package job;

import ch.hsr.geohash.GeoHash;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.naming.Context;
import java.awt.*;

// 车辆心跳数据POJO
public class Vehicle_a {
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class VehicleHeartbeat {
        private String vin;         // 车辆唯一标识
        private Long timestamp;     // 心跳时间戳
        private Double batteryLevel; // 电池电量
    }

    // 电池健康数据POJO
    @Data
    @AllArgsConstructor
    public class BatteryHealth {
        private String vin;
        private Double healthScore; // 0-100评分
    }

    @Data
    @AllArgsConstructor
    public class VehicleLocation {
        private String vin;
        private Double latitude;
        private Double longitude;
    }
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 实时在线车辆统计
        DataStream<Integer> onlineCountStream = heartbeatStream
                .keyBy(VehicleHeartbeat::getVin)
                .process(new KeyedProcessFunction<String, VehicleHeartbeat, Integer>() {

                    private ValueState<Long> lastActiveState;

                    @Override
                    public void open(Configuration parameters) {
                        lastActiveState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("lastActive", Long.class));
                    }

                    @Override
                    public void processElement(
                            VehicleHeartbeat heartbeat,
                            Context ctx,
                            Collector<Integer> out) throws Exception {

                        // 更新最后活跃时间
                        lastActiveState.update(heartbeat.getTimestamp());

                        // 注册5分钟后的定时器
                        ctx.timerService().registerProcessingTimeTimer(
                                ctx.timerService().currentProcessingTime() + 300_000);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws  Exception {
                        // 检查是否超时（5分钟无心跳）
                        if (lastActiveState.value() != null &&
                                timestamp - lastActiveState.value() >= 300_000) {
                            out.collect(-1);  // 车辆下线
                        } else {
                            out.collect(1);   // 车辆在线
                        }
                    }
                })
                .windowAll(TumblingProcessingTimeWindows.of(Time.minutes(5)))
                .aggregate(new AggregateFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    @Override
                    public Integer add(Integer value, Integer accumulator) {
                        return accumulator + value;
                    }

                    @Override
                    public Integer getResult(Integer accumulator) {
                        return Math.max(0, accumulator); // 避免负数
                    }

                    @Override
                    public Integer merge(Integer a, Integer b) {
                        return a + b;
                    }
                });

        // 每小时平均健康度计算
        DataStream<Double> avgBatteryHealthStream = batteryHealthStream
                .keyBy(BatteryHealth::getVin)
                .window(SlidingProcessingTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .aggregate(new AggregateFunction<BatteryHealth, Tuple2<Double, Integer>, Double>() {
                    @Override
                    public Tuple2<Double, Integer> createAccumulator() {
                        return Tuple2.of(0.0, 0);
                    }

                    @Override
                    public Tuple2<Double, Integer> add(BatteryHealth value, Tuple2<Double, Integer> acc) {
                        return Tuple2.of(acc.f0 + value.getHealthScore(), acc.f1 + 1);
                    }

                    @Override
                    public Double getResult(Tuple2<Double, Integer> acc) {
                        return acc.f1 == 0 ? 0 : acc.f0 / acc.f1;
                    }

                    @Override
                    public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, Tuple2<Double, Integer> b) {
                        return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
                    }
                });

        // 位置数据POJO


        // 地理热力图生成
        DataStream<GeoDensity> heatmapStream = locationStream
                .keyBy(loc -> GeoHash.encode(loc.getLatitude(), loc.getLongitude(), 6)) // 6位GeoHash约0.34km²
                .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
                .process(new ProcessWindowFunction<VehicleLocation, GeoDensity, String, TimeWindow>() {
                    @Override
                    public void process(
                            String geohash,
                            Context ctx,
                            Iterable<VehicleLocation> elements,
                            Collector<GeoDensity> out) {

                        int count = 0;
                        for (VehicleLocation ignored : elements) count++;

                        out.collect(new GeoDensity(
                                geohash,
                                ctx.window().getEnd(),
                                count,
                                GeoHash.decode(geohash).getCenterPoint()
                        ));
                    }
                });

        // 热力图数据POJO

    }

    // 模拟电池健康数据流（需根据实际数据源替换）
    DataStream<BatteryHealth> batteryHealthStream = env.fromElements(
            new BatteryHealth("VIN001", 95.5),
            new BatteryHealth("VIN002", 87.2)
            // 添加更多测试数据或替换为真实数据源
    );

    // 每小时平均健康度计算
    DataStream<Double> avgBatteryHealthStream = batteryHealthStream
            .keyBy(BatteryHealth::getVin)
            .window(SlidingProcessingTimeWindows.of(Time.hours(1), Time.minutes(5)))
            .aggregate(new AggregateFunction<BatteryHealth, Tuple2<Double, Integer>, Double>() {
                @Override
                public Tuple2<Double, Integer> createAccumulator() {
                    return Tuple2.of(0.0, 0);
                }

                @Override
                public Tuple2<Double, Integer> add(BatteryHealth value, Tuple2<Double, Integer> acc) {
                    return Tuple2.of(acc.f0 + value.getHealthScore(), acc.f1 + 1);
                }

                @Override
                public Double getResult(Tuple2<Double, Integer> acc) {
                    return acc.f1 == 0 ? 0 : acc.f0 / acc.f1;
                }

                @Override
                public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, Tuple2<Double, Integer> b) {
                    return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
                }
            });




    @Data
    @AllArgsConstructor
    public class GeoDensity {
        private String geohash;
        private Long timestamp;
        private Integer vehicleCount;
        private Point centerPoint; // 包含经纬度
    }
}
